package com.vincent.fk;

import java.util.*;
import java.util.concurrent.*;

/**
 * @author vincent
 * @DATE 2018/11/30
 * @Description
 */
public class ForkJoinWork extends RecursiveTask<Result> {
    private Task task;

    public ForkJoinWork(Task task) {
        this.task = task;
    }

    protected Result compute() {
        if (task.isSub()) {
            Result rs = new Result();
            // work
            try {
                System.out.println("begin run task, id="+task.getId());
                Thread.sleep(1000L);
                System.out.println("run task with condition="+task.getCondition());
                if (task.getCondition().contains("hehe")) {
                    throw new RuntimeException("不能呵呵");
                }
                rs.setSuccess(true);
                rs.setData(Collections.singletonList(UUID.randomUUID().toString()));
            } catch (Exception e) {
                rs.setSuccess(false);
                rs.setErrorMsg(e.getMessage());
            }
            return rs;
        } else {
            // join
            Result rs = null;
            // 拆分出子任务
            String[] split = task.getCondition().split(",");
            List<Task> taskList = new ArrayList<>(split.length);
            for (String cn : split) {
                Task sub = new Task();
                sub.setId(UUID.randomUUID().toString());
                sub.setSub(true);
                sub.setCondition(cn);
                taskList.add(sub);
            }
            task.setSubTaskList(taskList);
            // 提交子任务
            List<Result> results = new ArrayList<>(task.getSubTaskList().size());
            List<ForkJoinWork> tl = new ArrayList<>();
            for (Task task1 : task.getSubTaskList()) {
                ForkJoinWork forkJoinWork = new ForkJoinWork(task1);
                tl.add(forkJoinWork);
            }
            Collection<ForkJoinWork> forkJoinWorks = invokeAll(tl);
            for (ForkJoinWork forkJoinWork : forkJoinWorks) {
                try {
                    Result result = forkJoinWork.get(2, TimeUnit.SECONDS);
                    results.add(result);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }

            // 汇聚数据
            boolean isSuccess = true;
            rs = new Result();
            rs.setId(task.getId());
            for (Result result : results) {
                if (result.isSuccess()) {
                    List<String> data = rs.getData();
                    if (data == null) {
                        data = new LinkedList<>();
                        rs.setData(data);
                    }
                    data.addAll(result.getData());
                } else {
                    rs.setErrorMsg(result.getErrorMsg());
                    isSuccess = false;
                    break;
                }
            }
            rs.setSuccess(isSuccess);
            return rs;
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ForkJoinPool forkJoinPool = new ForkJoinPool(5);
        long start = System.currentTimeMillis();
        Task task = new Task();
        task.setId(UUID.randomUUID().toString());
        task.setSub(false);
        task.setCondition("haha,xixi,hehe,666");
        ForkJoinTask<Result> submit = forkJoinPool.submit(new ForkJoinWork(task));
        Result result = submit.get(2, TimeUnit.SECONDS);
        System.out.println("==============> 结果："+result.toString());
        long end = System.currentTimeMillis();
        System.out.println("cost :" + (end -start));
    }
}
